package com.loong.spark.streaming

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, SQLException}
import java.util.concurrent.LinkedBlockingQueue

import io.netty.handler.codec.string.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

/**
  * Created by 梁浩峰 on 2016/11/27 11:10.
  */
class ADClickedStreamingState {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("ADClickedStreamingState")
      /*.setJars(List(
      "",""
    ))*/
      .set("spark.streaming.receiver.writeheadLog.enable", "true")
      .set("spark.serializer", "org.apache.spark.serializer.KroSerializer")
      .set("spark.streaming.blockInterval", "100")
    val ssc = new StreamingContext(conf, Seconds(10))

    ssc.checkpoint("D:/sparkstreaingapps")

    val kafkaParameters = Map[String, String]("metadata.borker.list" -> "work01:9092,work02:9092,work03:9092")

    val topics = Set[String]("ADClickedStreamingState")

    val adclickedStreaming = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParameters, topics)


    val pairs = adclickedStreaming.map(t => {
      val splited = t._2.split("\t")
      val timestamp = splited(0)
      val ip = splited(1)
      val userId = splited(2)
      val adId = splited(3)
      val province = splited(4)
      val city = splited(5)
      val clickedRecord = timestamp + "_" + ip + "_" + userId + "_" + adId + "_" + province + "_" + city
      (clickedRecord, 1L)
    })

    val adClickedUsers = pairs.reduceByKey(_ + _)

    val filteredClickInBatch = adClickedUsers.filter(v => {
      if (1 < v._2) {
        false
      } else {
        true
      }
    })

    filteredClickInBatch.foreachRDD(rdd => {
      if (rdd.isEmpty()) {}
      rdd.foreachPartition(partition => {
        val userClickedList = ListBuffer[UserAdClicked]()
        while (partition.hasNext) {
          val record = partition.next()
          val splited = record._1.split("_")
          val userAdClicked = new UserAdClicked()
          userAdClicked.timestamp = splited(0)
          userAdClicked.ip = splited(1)
          userAdClicked.userId = splited(2)
          userAdClicked.adId = splited(3)
          userAdClicked.province = splited(4)
          userAdClicked.city = splited(5)
          userClickedList += userAdClicked
        }
        val inserting = ListBuffer[UserAdClicked]()
        val updating = ListBuffer[UserAdClicked]()
        val jdbcWrapper = JDBCWrapper.getInstance()

        for (clicked <- userClickedList) {
          def clickedquerycallBack(result: ResultSet) = {
            while (result.next()) {
              if ((result.getRow - 1) != 0) {
                val count = result.getLong(1)
                clicked.clickedCount = count
                updating += clicked
              } else {
                clicked.clickedCount = 0L
                inserting += clicked
              }
            }
          }

          jdbcWrapper.doQuery("SELECT count(1) FROM adclicked WHERE  timestamp = ? AND userId = ? AND adId = ?", Array(clicked.timestamp, clicked.userId, clicked.adId ), clickedquerycallBack
          )
        }

        val insertParametersList = ListBuffer[paramsList]()

        for (insertRecord <- inserting) {
          val paramsListTmp = new paramsList()
          paramsListTmp.params1 = insertRecord.timestamp
          paramsListTmp.params2 = insertRecord.ip
          paramsListTmp.params3 = insertRecord.userId
          paramsListTmp.params4 = insertRecord.adId
          paramsListTmp.params5 = insertRecord.province
          paramsListTmp.params6 = insertRecord.city
          paramsListTmp.params10_Long = insertRecord.clickedCount
          paramsListTmp.params_Type = "adclickedInsert"
          insertParametersList += paramsListTmp
        }

        jdbcWrapper.doBatch("INSERT INTO adclicked VALUES(?,?,?,?,?,?,?)", insertParametersList)


        val updateParametersList = ListBuffer[paramsList]()

        for (updateRecord <- updating) {
          val paramsListTmp = new paramsList()
          paramsListTmp.params1 = updateRecord.timestamp
          paramsListTmp.params2 = updateRecord.ip
          paramsListTmp.params3 = updateRecord.userId
          paramsListTmp.params4 = updateRecord.adId
          paramsListTmp.params5 = updateRecord.province
          paramsListTmp.params6 = updateRecord.city
          paramsListTmp.params10_Long = updateRecord.clickedCount
          paramsListTmp.params_Type = "adclickedUpdate"
          updateParametersList += paramsListTmp
        }

        jdbcWrapper.doBatch("UPDATE adclicked SET clickedCount = ? WHERE timestamp = ? AND ip = ? AND userId = ? AND adId =? AND provice =? AND city = ?", updateParametersList)

      })
    })



    ssc.start()
    ssc.awaitTermination()

  }

}


class UserAdClicked() {
  var timestamp: String = _
  var ip: String = _
  var userId: String = _
  var adId: String = _
  var province: String = _
  var city: String = _
  var clickedCount: Long = _
}


object JDBCWrapper {
  private var jdbcInstance: JDBCWrapper = _

  def getInstance(): JDBCWrapper = {
    synchronized {
      if (jdbcInstance == null) {
        jdbcInstance = new JDBCWrapper()
      }
      jdbcInstance
    }
  }
}

class JDBCWrapper {
  val dbConnectionPool = new LinkedBlockingQueue[Connection]()
  try {
    Class.forName("com.mysql.jdbc.Driver")
  } catch {
    case e: ClassNotFoundException => e.printStackTrace()
  }

  for (i <- 1 to 10) {
    try {
      val conn = DriverManager.getConnection("jdbc:mysql://master:3306/spark", "", "")
      dbConnectionPool.put(conn)
    } catch {
      case e: Exception => e.printStackTrace()
    }
  }

  def getConnection(): Connection = synchronized {
    while (0 == dbConnectionPool.size()) {
      try {
        Thread.sleep(20)
      } catch {
        case e: Exception => e.printStackTrace()
      }
    }
    dbConnectionPool.poll()
  }


  def doBatch(sqlText: String, paramsList: ListBuffer[paramsList]): Array[Int] = {
    val conn = getConnection()
    var ps: PreparedStatement = null
    var rs: Array[Int] = null
    try {
      conn.setAutoCommit(false)
      ps = conn.prepareStatement(sqlText)
      for (parameters <- paramsList) {
        parameters.params_Type match {
          case "adclickedInsert" => {
            println("adclickedInsert")

            ps.setObject(1, parameters.params1)
            ps.setObject(2, parameters.params2)
            ps.setObject(3, parameters.params3)
            ps.setObject(4, parameters.params4)
            ps.setObject(5, parameters.params5)
            ps.setObject(6, parameters.params6)
            ps.setObject(7, parameters.params10_Long)
          }

          case "blacklisttableInsert" => {
            println("blacklisttableInsert")
            ps.setObject(1, parameters.params1)
          }
          case "adclickedcountInsert" => {
            println("adclickedcountInsert")
            ps.setObject(1, parameters.params1)
            ps.setObject(2, parameters.params2)
            ps.setObject(3, parameters.params3)
            ps.setObject(4, parameters.params4)
            ps.setObject(5, parameters.params10_Long)
          }
          case "adprovincetopnInsert" => {
            println("adprovincetopnInsert")
            ps.setObject(1, parameters.params1)
            ps.setObject(2, parameters.params2)
            ps.setObject(3, parameters.params3)
            ps.setObject(4, parameters.params10_Long)
          }
          case "adclickedtrendInsert" => {
            println("adclickedtrendInsert")
            ps.setObject(1, parameters.params1)
            ps.setObject(2, parameters.params2)
            ps.setObject(3, parameters.params3)
            ps.setObject(4, parameters.params4)
            ps.setObject(5, parameters.params10_Long)
          }
          case "adclickedUpdate" => {
            println("adclickedUpdate")
            ps.setObject(1, parameters.params10_Long)
            ps.setObject(2, parameters.params1)
            ps.setObject(3, parameters.params2)
            ps.setObject(4, parameters.params3)
            ps.setObject(5, parameters.params4)
            ps.setObject(6, parameters.params5)
            ps.setObject(7, parameters.params6)

          }

          case "blacklisttableUpdate" => {
            println("blacklisttableUpdate")
            ps.setObject(1, parameters.params1)
          }
          case "adclickedcountUpdate" => {
            println("adclickedcountUpdate")
            ps.setObject(1, parameters.params10_Long)
            ps.setObject(2, parameters.params1)
            ps.setObject(3, parameters.params2)
            ps.setObject(4, parameters.params3)
            ps.setObject(5, parameters.params4)

          }
          case "adprovincetopnUpdate" => {
            println("adprovincetopnUpdate")
            ps.setObject(1, parameters.params10_Long)
            ps.setObject(2, parameters.params1)
            ps.setObject(3, parameters.params2)
            ps.setObject(4, parameters.params3)

          }

          case "adprovincetopnDelete" => {
            println("adprovincetopnDelete")

            ps.setObject(1, parameters.params1)
            ps.setObject(2, parameters.params2)

          }

          case "adclickedtrendUpdate" => {
            println("adclickedtrendUpdate")
            ps.setObject(1, parameters.params10_Long)
            ps.setObject(2, parameters.params1)
            ps.setObject(3, parameters.params2)
            ps.setObject(4, parameters.params3)
            ps.setObject(5, parameters.params4)
          }
            ps.addBatch()
        }
      }
      rs = ps.executeBatch()
      conn.commit();
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      if (ps != null) {
        try {
          ps.close()
        } catch {
          case e: SQLException => e.printStackTrace()
        }
      }

      if (conn != null) {
        try {
          dbConnectionPool.put(conn)
        } catch {
          case e: InterruptedException => e.printStackTrace()
        }
      }
    }
    rs
  }


  def doQuery(sqlText: String, paramsList: Array[_], callBack: ResultSet => Unit) = {
    val conn = getConnection()
    var ps: PreparedStatement = null
    var rs: ResultSet = null
    try {
      ps = conn.prepareStatement(sqlText)
      if (paramsList != null) {
        for (i <- 0 to paramsList.length - 1) {
          ps.setObject(i + 1, paramsList(i))
        }
      }
      rs = ps.executeQuery()
      callBack(rs)
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      if (ps != null) {
        try {
          ps.close();
        } catch {
          // TODO Auto-generated catch block
          case e: SQLException => e.printStackTrace()

        }
      }

      if (conn != null) {
        try {
          dbConnectionPool.put(conn);
        } catch {
          case e: InterruptedException => e.printStackTrace()
        }
      }
    }
    rs
  }

  def resultCallBack(result: ResultSet, blackListNames: List[String]): Unit = {

  }
}


class paramsList extends Serializable {
  var params1: String = _
  var params2: String = _
  var params3: String = _
  var params4: String = _
  var params5: String = _
  var params6: String = _
  var params7: String = _
  var params10_Long: Long = _
  var params_Type: String = _
  var length: Int = _

}






